package com.upcoming.data.storm.fileRead;

import com.upcoming.data.storm.util.MyUtil;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

/**
 * 作者：tianjiayuan
 * 创建时间：2018-05-23 10:32
 * 类描述：
 * 修改人：
 * 修改时间：
 */
public class WordBolt extends BaseRichBolt {

    private OutputCollector collector;

    private Map<String, Integer> wordCountMap = new HashMap<>();

    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
    }

    public void execute(Tuple tuple) {
        String word = tuple.getString(0);
        MyUtil.outLog2NC(this, "execute:" + word);
        if (wordCountMap.containsKey(word)) {
            wordCountMap.put(word, wordCountMap.get(word) + 1);
        } else {
            wordCountMap.put(word, 1);
        }
        this.collector.ack(tuple);
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    public WordBolt() {
        super();
    }

    public void cleanup() {
        for (Map.Entry<String, Integer> entry : wordCountMap.entrySet()) {
            MyUtil.outLog2NC(this, entry.getKey() + ":" + entry.getValue());
        }
        super.cleanup();
    }

    public Map<String, Object> getComponentConfiguration() {
        return super.getComponentConfiguration();
    }
}
